package com.autodesk.akn.emr.util;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.util.Properties;

import org.apache.commons.httpclient.HostConfiguration;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
import org.springframework.stereotype.Repository;

import com.autodesk.akn.emr.services.S3Service;
import com.autodesk.akn.emr.util.s3nativefilesystem.Jets3tNativeFileSystemStore;
import com.autodesk.akn.emr.util.s3nativefilesystem.NativeFileSystemStore;
import com.autodesk.akn.emr.util.s3nativefilesystem.NativeS3FileSystem;

@Repository
public class S3Util {
	private final static Logger LOGGER = Logger.getLogger(S3Util.class
			.getName());
	private NativeS3FileSystem fs;

	public S3Util() {
		try {
			setFileSystem();
		} catch (Exception e) {
			throw new RuntimeException("Error intializing S3 file system");
		}
	}

	public void setFileSystem() throws IOException {
		// Obtain the access key and the secret key from the property file
		Properties prop = new Properties();
		prop.load(S3Service.class.getClassLoader().getResourceAsStream(
				"config.properties"));
		String accessKey = (String) prop.get("credentials.accessKey");
		String secretKey = (String) prop.get("credentials.secretKey");
		String proxy = (String) prop.get("proxy.enabled");
		boolean proxyneeded = "true".equals(proxy);
		// Add the keys to the conf object and pass it to the file system
		Configuration conf = new Configuration();
		conf.set("fs.s3n.awsAccessKeyId", accessKey);
		conf.set("fs.s3n.awsSecretAccessKey", secretKey);
		URI uri = URI.create("s3n://com.autodesk.akn/");
		HostConfiguration hostConfiguration = new HostConfiguration();
		if (proxyneeded) {
			hostConfiguration.setProxy("proxy.corp.globant.com", 3128);
		}
		NativeFileSystemStore store = new Jets3tNativeFileSystemStore();
		fs = new NativeS3FileSystem(store);
		fs.initialize(uri, conf, hostConfiguration);
		LOGGER.info("File system detected and initialized");
	}

	public String copySingleFileFromS3(String inputBucket, String inputKey) {
		String[] splittedName = inputKey.split("/");
		String fileName = splittedName[splittedName.length - 1];
		FSDataInputStream s3InputStream = null;
		FileOutputStream localFileOutputStream = null;
		try {
			Path inFile = new Path("s3n://" + inputBucket + "/" + inputKey);
			s3InputStream = fs.open(inFile);
			File inputFile = File.createTempFile(fileName, "");
			localFileOutputStream = new FileOutputStream(inputFile);

			IOUtils.copy(s3InputStream, localFileOutputStream);
			fileName = inputFile.getPath();
			LOGGER.info("Copied file: " + fileName);
		} catch (IOException e) {
			LOGGER.error(e.getMessage());
		} finally {
			try {
				if (s3InputStream != null)
					s3InputStream.close();
				if (localFileOutputStream != null)
					localFileOutputStream.close();
			} catch (IOException e) {
				LOGGER.error(e.getMessage());
			}
		}
		return fileName;
	}

	public int moveFiles(String from, String to) {
		FSDataInputStream s3InputStream = null;
		FSDataOutputStream s3OutputStream = null;
		int size = 0;
		try {
			Path inFile = new Path(from);
			Path outFile = new Path(to);
			s3InputStream = fs.open(inFile);
			s3OutputStream = fs.create(outFile);
			size = IOUtils.copy(s3InputStream, s3OutputStream);
			fs.delete(inFile, false);
		} catch (IOException e) {
			LOGGER.error(e.getMessage());
		} finally {
			try {
				if (s3InputStream != null)
					s3InputStream.close();
				if (s3OutputStream != null)
					s3OutputStream.close();
			} catch (IOException e) {
				LOGGER.error(e.getMessage());
			}
		}
		return size;
	}

	/**
	 * Clean the workspace directory
	 */
	public void cleanWorkspaceDirectory() {
		try {
			Path extraInfo = new Path("s3n://com.autodesk.akn/ProcessedLogs/ExtraInformation");
			Path toDatabase = new Path("s3n://com.autodesk.akn/ProcessedLogs/toDatabase");
			Path logsProcessed = new Path("s3n://com.autodesk.akn/ProcessedLogs/LogsProcessed");
			if (fs.exists(extraInfo)) {
				fs.delete(extraInfo, true);	
			}
			if (fs.exists(toDatabase)) {
				fs.delete(toDatabase, true);
			}
			if (fs.exists(logsProcessed)) {
				fs.delete(logsProcessed, true);
			}
		} catch (IOException e) {
			LOGGER.error(e.getMessage());
		}
	}
}
